package com.lvmama.pangolin.kafka;


import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Lists;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.log4j.Logger;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author wuchuantong
 * @version V1.0
 * @Title: KafkaConsumerTemplate
 * @Package com.lvtu.client.push.jms
 * @Description: kafka消费者模版类
 * @date 2016/12/26 15:32
 */
public abstract class KafkaConsumerTemplate<T> {

    private final Logger logger = Logger.getLogger(this.getClass());


    private Properties properties;
    private Integer partition;
    private List<String> topics;
    private Integer pollTimeout = 200;
    private Deserializer keyDeserializer = new StringDeserializer();
    private Deserializer valueDeserializer = new ByteArrayDeserializer();
    //限流功能，为0时表示不限制，大于0时，表示限制所有消费者每秒能处理的消息总数
    private Integer speed = 0;
    //限流计数器
    private LoadingCache<Long, AtomicLong> counter =
            CacheBuilder.newBuilder()
                    .expireAfterWrite(2, TimeUnit.SECONDS)
                    .build(new CacheLoader<Long, AtomicLong>() {
                        @Override
                        public AtomicLong load(Long seconds) throws Exception {
                            return new AtomicLong(0);
                        }
                    });


    public KafkaConsumerTemplate(Properties properties, Integer partition, List<String> topics, Integer pollTimeout, Deserializer keyDeserializer, Deserializer valueDeserializer, Boolean runOnlyMaster) {
        this.properties = properties;
        this.partition = partition;
        this.topics = topics;
        if (pollTimeout != null) {
            this.pollTimeout = pollTimeout;
        }
        if (keyDeserializer != null) {
            this.keyDeserializer = keyDeserializer;
        }
        if (valueDeserializer != null) {
            this.valueDeserializer = valueDeserializer;
        }
    }

    public KafkaConsumerTemplate(Properties properties, Integer partition, List<String> topics, Integer pollTimeout) {
        this(properties, partition, topics, pollTimeout, null, null, null);
    }

    public KafkaConsumerTemplate(Properties properties, Integer partition, List<String> topics) {
        this(properties, partition, topics, null, null, null, null);
    }

    public KafkaConsumerTemplate() {
    }

    public void startConsumer() {
        logger.info("startConsumer...");
        if (partition == null) {
            throw new RuntimeException("please config param partition");
        }
        if (partition < 1) {
            throw new RuntimeException("param threadCount must more than 1");
        }
        if (CollectionUtils.isEmpty(topics)) {
            throw new RuntimeException("please config param topics");
        }

        //设置处理消息线程数，线程数应小于等于partition数量，若线程数大于partition数量，则多余的线程则闲置，不会进行工作
        //   final Properties props = ConfigHelper.getProperties("consumer.properties");
        //声明一个线程池，用于消费各个partition
        ExecutorService executor = Executors.newFixedThreadPool(partition);
        //创建指定的线程消费
        for (int i = 0; i < partition; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {

                    KafkaConsumer<String, byte[]> consumer = null;
                    try {
                        consumer = new KafkaConsumer<String, byte[]>(properties, keyDeserializer, valueDeserializer);
                        consumer.subscribe(Lists.newArrayList(topics));
                        while (true) {
                            ConsumerRecords<String, byte[]> records = consumer.poll(pollTimeout);
                            for (ConsumerRecord<String, byte[]> record : records) {
                                T message = null;
                                try {
                                    //将字节码反序列化成相应的对象
                                    byte[] bytes = record.value();
                                    message = (T) SerializationUtils.deserialize(bytes);
                                    logger.info("consumed " + record.partition() + "th ,message with offset: " + record.offset() + ", value:" + message);
                                } catch (Exception e) {
                                    logger.error(record.topic() + " consumer error:" + e.getMessage());
                                    logger.error(e.getMessage(), e);
                                    return;
                                }

                                //判断是否启动限速功能
                                if (speed > 0) {
                                    while (true) {
                                        //得到当前秒数
                                        long currentSeconds = System.currentTimeMillis() / 1000;
                                        if (counter.get(currentSeconds).incrementAndGet() > speed) {
                                            logger.info("maximum speed reached,currentSeconds:" + currentSeconds + ",speed:" + speed);
                                            Thread.sleep(20);
                                            continue;
                                        }
                                        //
                                        break;
                                    }
                                }
                                //调用自己的业务逻辑
                                try {
                                    process(message);
                                } catch (Exception e) {
                                    logger.error(record.topic() + " consumer handle error:" + e.getMessage());
                                    logger.error(e.getMessage(), e);
                                }

                            }
                        }
                    } catch (Exception e) {
                        logger.error("init consumer error:" + e.getMessage());
                        logger.error(e.getMessage(), e);
                    }
                }
            });
        }
    }

    abstract void process(T message);


    public Properties getProperties() {
        return properties;
    }

    public void setProperties(Properties properties) {
        this.properties = properties;
    }

    public Integer getPartition() {
        return partition;
    }

    public void setPartition(Integer partition) {
        this.partition = partition;
    }

    public List<String> getTopics() {
        return topics;
    }

    public void setTopics(List<String> topics) {
        this.topics = topics;
    }

    public Integer getPollTimeout() {
        return pollTimeout;
    }

    public void setPollTimeout(Integer pollTimeout) {
        this.pollTimeout = pollTimeout;
    }

    public Deserializer getKeyDeserializer() {
        return keyDeserializer;
    }

    public void setKeyDeserializer(Deserializer keyDeserializer) {
        this.keyDeserializer = keyDeserializer;
    }

    public Deserializer getValueDeserializer() {
        return valueDeserializer;
    }

    public void setValueDeserializer(Deserializer valueDeserializer) {
        this.valueDeserializer = valueDeserializer;
    }


    public Integer getSpeed() {
        return speed;
    }

    public void setSpeed(Integer speed) {
        this.speed = speed;
    }
}
